package com.cnlive.logpare.app
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import com.cnlive.logpare.util.IPSeekerExt
import scala.collection.mutable.HashMap
import java.net.URLDecoder
import com.cnlive.logpare.util.ParseLogUtil
import org.apache.spark.sql.Row
import java.text.SimpleDateFormat
import java.util.Locale
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.types.IntegerType
import org.apache.spark.sql.types.LongType
import org.apache.spark.sql.types.StringType
import org.apache.spark.sql.SQLContext
import java.util.Properties
import java.sql.DriverManager

object ParseAppLogPhnix {
  val sdf=new SimpleDateFormat("dd/MMM/yyyy:HH:mm:ss",Locale.ENGLISH)
  val dateFormat=new SimpleDateFormat("yyyyMMddHHmmss",Locale.ENGLISH)
  val matcheLine="""([^ ]*) ([^ ]*) ([^ ]*) \[(.*)\] \"GET(.*?)\" (-|[0-9]*) (-|[0-9]*) \"(.*?)\" \"(.*?)\""""
  def main(args: Array[String]): Unit = {
  val env=1
    val conf = new SparkConf().setAppName("NewParsePageLog")
    if(env==1){
      conf.setMaster("local")
    }
    val sc = new SparkContext(conf)
    val files=env match{
      case 1 => "file:///F:/工作资料/app/*"
      case 2 => "hdfs://sxt/spark/test/input/logs/page/*"
      case 3 => "hdfs://sxt/nginx_logs/events//*"
    }
    val qqwryPath=env match{
          case 2 =>"/home/yarn/qqwry.dat"
          case 3 =>"/home/dwsj/had/hadoop-2.6.5/lib/qqwry.dat"
          case _=>""
    }
    val zk_addr=env match{
        case 3=>"node001,node002,node003:2181"
        case _=>"node180,node181,node182:2181"
      }
  /* val dd= sc.textFile(files).filter { x => x.matches(matcheLine) }.map(x => (x,1)).reduceByKey(_+_)
  val f= dd.filter(x=>x._2>1).foreach(println(_))*/
   val tabData= sc.textFile(files).mapPartitions(x=> {
      val props=new Properties;
      val conn = DriverManager.getConnection("jdbc:phoenix:node180,node181,node181:2181", props);
      val d=conn.createStatement()
      var result = List[Row]()
      val sessionMap=new HashMap[String, Int]()
      val iPSeekerExt=env match {
          case 1 =>  IPSeekerExt.getInit()
          case 2 =>  IPSeekerExt.getInit(true,qqwryPath)
          case 3 =>  IPSeekerExt.getInit(true,qqwryPath)
      }
      while (x.hasNext) {
        val lineText=x.next()
        if(lineText.matches(matcheLine)){
          result= result.::(parLineText(iPSeekerExt,lineText,sessionMap))
        }
      }
      result.iterator
    }).filter(x => x.getInt(0)==1)
    //.take(10).foreach(println(_))
    /*val schema = StructType(Seq(
        StructField("ISFILTER", IntegerType, nullable = true),
        StructField("NGINXTIME", LongType, nullable = true),
        StructField("TM", LongType, nullable = true),
        StructField("UID", StringType, nullable = true),
        StructField("SPID", StringType, nullable = true),
        StructField("APPID", StringType, nullable = true),
        StructField("IP", StringType, nullable = true),
        StructField("VERSION", StringType, nullable = false),
        StructField("CITY", StringType, nullable = false),
        StructField("LOCATION", StringType, nullable = false),
        StructField("OPERATOR", IntegerType, nullable = false),
        StructField("PLAT", IntegerType, nullable = false),
        StructField("EVENTID", IntegerType, nullable = false),
        StructField("ISNEWUSER", IntegerType, nullable = true),
        StructField("ISSTART", IntegerType, nullable = true)
      ))
     val sqlContext = new SQLContext(sc)
     import org.apache.phoenix.spark._
     val df=  sqlContext.createDataFrame(tabData, schema)
     df.registerTempTable("tempTb")
     val dd= sqlContext.sql("""select 
            NGINXTIME,TM,UID,SPID,APPID,IP,VERSION,CITY,
            LOCATION,OPERATOR,PLAT,EVENTID,ISNEWUSER,ISSTART
            from tempTb""")
            //.count()
           //.println(dd)
            //3571 3491 3562  4 3568
      // .foreach { println(_) }
     dd.saveToPhoenix("APPLOG", zkUrl=Some(zk_addr))*/
  }
  
  
  
  
  
  
  
  
  
  
  
  val ts=""" \"(.*?)\"""".r
  def parLineText(iPSeekerExt:IPSeekerExt,lineText:String,sessionMap:HashMap[String, Int]):Row={
     val mapRequest=new HashMap[String, String]()
     val ipandtime=ts.split(lineText).toList
     val urlandtype=ts.findAllIn(lineText).toList
     if(ipandtime.length==2&&urlandtype.length==3){//
      val ipandtimeArr=ipandtime(0).split(" ")
      if(ipandtimeArr.length==5){
        val ip=ipandtimeArr(0)
        val time=ipandtimeArr(3).replace("[", "")
        mapRequest.put("nginxtime", time)//时间
        val d=iPSeekerExt.analyticIp(ip)
        val city=d.getCity
        val province=d.getProvince
        val country=d.getCountry
        val operator=d.getOperator
        mapRequest.put("ip", ip);//城市
        mapRequest.put("location", province);//省份
        mapRequest.put("city", city);//城市
        mapRequest.put("operator", operator+"")//运营商
      }else{
        println("时间跟ip解析有误"+lineText)
      }
       val urlandper=urlandtype(0).split(" ")
       if(urlandper.length==4){//解析url
          URLRequest(urlandper(2),mapRequest,sessionMap)
       }else{
        println("url解析有误"+lineText)
       }
       platParse(urlandtype(2),mapRequest)//解析操作系统
     }else{//数据格式不对
       println("数据格式不对"+lineText)
     }
     if(mapRequest.contains("appid")
         &&mapRequest.contains("nginxtime")
         &&mapRequest.contains("eventid")
         &&mapRequest.contains("ip")
         &&mapRequest.contains("tm")
         &&mapRequest.contains("uid")
         &&mapRequest.contains("location")//省份
         &&mapRequest.contains("plat")//操作系统
         &&mapRequest.contains("city")//城市
         &&mapRequest.contains("operator")//运营商
         &&mapRequest.contains("isnewuser")//是否新用户
         ){
       val appid=mapRequest.getOrElse("appid", "null")
       val spidandappid=appid.split("_")
       val nginxtime=dateFormat.format(sdf.parse(mapRequest.getOrElse("nginxtime", "20101010000000"))).toLong
       val city=mapRequest.getOrElse("city", "null")
       val location=mapRequest.getOrElse("location", "null")
       val plat=mapRequest.getOrElse("plat", "199").toInt
       val eventid=mapRequest.getOrElse("eventid", "0").toInt
       val version=mapRequest.getOrElse("version", "0")
       val uid=mapRequest.getOrElse("uid", "0")
       val ip=mapRequest.getOrElse("ip", "0")
       val tm=mapRequest.getOrElse("tm", "0").toLong
       val operator=mapRequest.getOrElse("operator", "0").toInt
       val isnewuser=mapRequest.getOrElse("isnewuser", "0").toInt
       val isstart=if(eventid==1300)1 else 0 
       if(spidandappid.length==2){
         //NGINXTIME  | UID  | SPID  | APPID  | IP  | VERSION  | CITY  | LOCATION  | OPERATOR  | PLAT  | EVENTID  | ISNEWUSER
         //println(nginxtime,tm,uid,spidandappid(0),appid,ip,version,city,location,operator,plat,eventid,isnewuser,isstart)
           Row(1,nginxtime,tm,uid,spidandappid(0),appid,ip,version,city,location,operator,plat,eventid,isnewuser,isstart)
       }else{
           println("appid不是_分割的")
           Row(0)
       }
     }else{
       // mapRequest.foreach(println(_))
        println("appid为空")
        Row(0)
     }
  }
  //解析操作系统
  def platParse(text:String,mapRequest:HashMap[String, String]){
    val d= ParseLogUtil.get_mobile_system(URLDecoder.decode(text,"utf-8").toLowerCase())
     mapRequest.put("plat",d+"")
  }
  //url解析
  def URLRequest(url:String,mapRequest:HashMap[String, String],sessionMap:HashMap[String, Int]){
   val strUrlParam=TruncateUrlPage(URLDecoder.decode(url,"utf-8").toLowerCase())
   if(strUrlParam.length!=0){
     val arrSplit=strUrlParam.split("&")
     for(strSplit<-arrSplit){
       val arrSplitEqual=strSplit.split("[=]")
       if(arrSplitEqual.length>1){
         mapRequest.put(arrSplitEqual(0), arrSplitEqual(1));
       }
     }
   }
   val uid=mapRequest.getOrElse("uid", "")
   val tm=mapRequest.getOrElse("tm", "")
   if((!uid.equals(""))&&(!tm.equals(""))){
     val uidAndtm=uid.split("_")
     if(uidAndtm(0).length()==13){//uid中包含1494210652000_j1tsu24521
       if(uidAndtm(0).toLong==tm.toLong){
         if(sessionMap.contains(uid)){
          mapRequest.put("isnewuser", "0");   
         }else{
            sessionMap.put(uid, 1);
            mapRequest.put("isnewuser", "1");   
         }
       }else{
          mapRequest.put("isnewuser", "0"); 
       }
     }else{
       try{
         if(uidAndtm(1).toLong==tm.toLong){
           if(sessionMap.contains(uid)){
             mapRequest.put("isnewuser", "0");   
           }else{
              sessionMap.put(uid, 1);
              mapRequest.put("isnewuser", "1");   
           }
         }else{
            mapRequest.put("isnewuser", "0"); 
         }
       }catch {
         case t: Throwable => mapRequest.put("isnewuser", "0");
       }
     }
   }else{
     println("url参数解析不合格！")
   }
  }
  //截取URL
  def TruncateUrlPage(strURL:String):String={
    val arrSplit=strURL.split("[?]")
    var strAllParam="";
    if(strURL.length()>1){
      if(arrSplit.length>1){
          if(arrSplit(1)!=null){
           strAllParam=arrSplit(1)
          }
      }
    }
    strAllParam
  }
}